package extract.helper;

import common.config.DbTableNames;
import common.dao.couchdb.CouchDBKGFactory;
import common.dao.mysql.MySqlDao;
import org.ektorp.CouchDbConnector;
import org.ektorp.changes.ChangesCommand;
import org.ektorp.changes.DocumentChange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by Common on 11/24/16.
 */

/*
1) http://ektorp.org/reference_documentation.html#d100e1039
2) http://docs.couchdb.org/en/2.0.0/api/database/changes.html#changes-longpoll
3) http://www.programcreek.com/java-api-examples/index.php?api=org.ektorp.changes.DocumentChange
 */

public class CouchdbChangeHelper {

    public static final Logger logger = LoggerFactory.getLogger(CouchdbChangeHelper.class);

    private CouchDbConnector couchdb;
    private MySqlDao mysqldao;
    private Connection conn;

    private DbTableNames.DBType db_type;
    private DbTableNames.TBType couchdb_type;

    private String db_name;
    private String tb_name;
    private String couchdb_name;



    public CouchdbChangeHelper(CouchDbConnector db, MySqlDao dao, DbTableNames.DBType db_type, DbTableNames.TBType couchdb_type){
        this.couchdb = db;
        this.mysqldao = dao;
        conn = this.mysqldao.getCon();
        this.db_type = db_type;
        this.couchdb_type = couchdb_type;
        this.db_name = DbTableNames.MySQLDB.DB_NAME;
        this.tb_name = DbTableNames.MySQLDB.getTableName( this.db_type );
        this.couchdb_name = DbTableNames.CouchDB.getCouchDBName(this.couchdb_type);
    }

    @Override
    protected void finalize(){
        try {
            super.finalize();
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        }
        this.mysqldao.closeCon();
    }



    public void insertNewestSeqNumber(String id, int serial, int status, String rev, String parent_rev, String parent_id){
        String sql = String.format("replace into %s.%s (couchdb,id,serial,status,rev,parent_rev,parent_id) values ('%s','%s',%d,%d,'%s','%s','%s');",
                this.db_name,this.tb_name,this.couchdb_name,id,serial,status,rev,parent_rev,parent_id );
        logger.info("[insertNewestSeqNumber] sql = {}", sql);
        try {
            Statement stmt = conn.createStatement();
            stmt.execute(sql);
            stmt.close();
        } catch (SQLException e){
            e.printStackTrace();
        }
    }

    public void insertNewestSeqNumber(String id, int serial, int status){
        String sql = String.format("replace into %s.%s (couchdb,id,serial,status) values ('%s','%s',%d,%d);",
                this.db_name,this.tb_name,this.couchdb_name,id,serial,status );
        logger.info("[insertNewestSeqNumber] sql = {}", sql);
        try {
            Statement stmt = conn.createStatement();
            stmt.execute(sql);
            stmt.close();
        } catch (SQLException e){
            e.printStackTrace();
        }

    }


    public List<String> getChildIDs(String parent_id){
        String sql = String.format("select ts,couchdb,id,serial,status from %s.%s where couchdb='%s' and status=1 and parent_id='%s'; ",
                this.db_name, this.tb_name, this.couchdb_name, parent_id);
        logger.info("[sql = ] {}", sql);
        List<String> ids = new ArrayList<>();
        try{
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);
            while(rs.next()){
                String id = rs.getString("id");
                ids.add(id);
            }

            stmt.close();
        }catch (Exception e){
            e.printStackTrace();
        }
        return ids;
    }


    public List<String> getChildIDsByParentRev(String parent_id, String parent_rev){
        String sql = String.format("select ts,couchdb,id,serial,status from %s.%s where couchdb='%s' and status=1 and parent_id='%s' and parent_rev<'%s'; ",
                this.db_name, this.tb_name, this.couchdb_name, parent_id, parent_rev);
        logger.info("[sql = ] {}", sql);
        List<String> ids = new ArrayList<>();
        try{
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);
            while(rs.next()){
                String id = rs.getString("id");
                ids.add(id);
            }

            stmt.close();
        }catch (Exception e){
            e.printStackTrace();
        }
        return ids;
    }


    public int getLastSeqNumberInMysql(){
        String sql = String.format("select ts,couchdb,id,serial,status from %s.%s where couchdb='%s' and (status=1 or status=-1) " +
                        "order by ts desc,serial desc limit 1;",
                this.db_name , this.tb_name, this.couchdb_name);
        logger.debug("[sql = ] {}", sql);
        int last_seq = -1;
        try {
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery(sql);
            int index = 0;
            while(rs.next()){
                //System.out.println(index);
                //System.out.println(rs);
                last_seq = rs.getInt("serial");
                index += 1;
            }
            stmt.close();

        } catch (SQLException e) {
            e.printStackTrace();
        }
        logger.info("[getLastSeqNumberInMysql] last_seq = {}", last_seq);
        return last_seq;
    }

    public int getFirstSeqNumberInCouchdb(){
        ChangesCommand cmd = new ChangesCommand.Builder()
                .limit(1)
                .param("descending","false")
                .build();
        List<DocumentChange> changes = this.couchdb.changes(cmd);
        int last_seq = -1;
        for(DocumentChange change: changes){
            last_seq = change.getSequence();
        }
        logger.info("[getLastSeqNumberInCouchdb] last_seq = {}", last_seq);
        return last_seq;
    }

    public int getLastSeqNumber(){
        int last_seq = getLastSeqNumberInMysql();
        if(last_seq == -1){
            //last_seq = getFirstSeqNumberInCouchdb();
            last_seq = 0;
        }
        logger.info("[getLastSeqNumber] last_seq = {}", last_seq);
        return last_seq;
    }

    public List<DocumentChange> getChanges(int start, int size, boolean include_doc){
        ChangesCommand cmd = new ChangesCommand.Builder()
                .since(start)
                .limit(size)
                .includeDocs(include_doc)
                .build();
        List<DocumentChange> changes = this.couchdb.changes(cmd);
        return changes;
    }

    public boolean isHaveChangeSinceSeqNum(int seqNum){
        logger.info("[isHaveChangeSinceSeqNum] seqNum = {}", seqNum);
        List<DocumentChange> changes = getChanges(seqNum, 1, false);
        boolean ret = false;
        if(changes.size() > 0)
            ret =  true;
        logger.info("[isHaveChangeSinceSeqNum] ret = {}", ret);
        return ret;
    }


    public boolean isHaveChange(){
        int lastSequenceNumber = getLastSeqNumber();
        logger.info("[isHaveChange] lastSequenceNumber = {}", lastSequenceNumber);
        return isHaveChangeSinceSeqNum( lastSequenceNumber );
    }


    /*
    *******************************************************
    main func
     */
    public static void main(String[] args){
        //CouchDbConnector concept_con = new CouchDBDOCFactory().getVideoCon();
        CouchDbConnector concept_con = new CouchDBKGFactory().getConceptCon();
        CouchdbChangeHelper helper = new CouchdbChangeHelper(concept_con, new MySqlDao(),
                DbTableNames.DBType.KG, DbTableNames.TBType.CONCEPT);

        //System.exit(0);

        int seq_num = helper.getLastSeqNumber(); //181364;
        //seq_num = 181364;
        boolean ret = helper.isHaveChangeSinceSeqNum( seq_num );
        System.out.println("ret = " + ret);

        //String couchdb_name = DbTableNames.KGCouchDB.CONCEPT_TB;
        int size = 10;
        List<DocumentChange> changes = helper.getChanges(seq_num, size, true);
        for(DocumentChange change : changes) {
            String id = change.getId();
            //new_id = id;
            int seq = change.getSequence();
            //new_serial = seq;
            String doc_json = change.getDoc();
            logger.info("id = {}, seq = {}\n", id, seq);
            logger.info("doc_json = \n{}" ,doc_json);

            helper.insertNewestSeqNumber(id,seq,1);
        }
    }



}
